package net.sina.realtime.traffic.job3;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.sina.realtime.traffic.utils.JdbcUtil;
import net.sina.realtime.traffic.utils.KafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class IllegalVehicleTrackJob {
    public static void main(String[] args) throws Exception {
        //1.执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.数据源-source
        DataStream<String> stringDataStream = KafkaUtil.consumerKafka(env, "traffic-log");


        //3.数据处理-transformation
//       DataStream<String> resultStream = hedal(stringDataStream);
//       resultStream.print("kafka");
        SingleOutputStreamOperator<String> resultStream = stringDataStream.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String eventType = jsonObject.getString("eventType");
                JSONObject jsonObject1 = new JSONObject();
                if (("normal".equals(eventType))) {
                    jsonObject.remove("event_id");
                    jsonObject.remove("camera_id");
                    jsonObject.remove("license_plate");
                    jsonObject.remove("vehicle_type");
                    jsonObject.remove("road_id");
                    jsonObject.remove("crossing_id");
                    jsonObject.remove("timestamp");
                    jsonObject.remove("speed");
                    jsonObject.remove("direction");
                    jsonObject.remove("lane");
                    jsonObject.remove("event_type");
                    jsonObject.remove("area_id");
                    return jsonObject.toJSONString();
                }else {
                    jsonObject1.put("eventId", jsonObject.getString("event_id"));
                    jsonObject1.put("cameraId", jsonObject.getString("camera_id"));
                    jsonObject1.put("licensePlate", jsonObject.getString("license_plate"));
                    jsonObject1.put("vehicleType", jsonObject.getString("vehicle_type"));
                    jsonObject1.put("roadId", jsonObject.getString("road_id"));
                    jsonObject1.put("crossingId", jsonObject.getString("crossing_id"));
                    jsonObject1.put("ts", jsonObject.getLong("timestamp"));
                    jsonObject1.put("speed", jsonObject.getDouble("speed"));
                    jsonObject1.put("direction", jsonObject.getString("direction"));
                    jsonObject1.put("lane", jsonObject.getIntValue("lane"));
                    jsonObject1.put("eventType", jsonObject.getString("event_type"));
                    jsonObject1.put("areaId", jsonObject.getString("area_id"));
                    return jsonObject1.toJSONString();
                }
            }
        });
//        SingleOutputStreamOperator<String> aNull = resultStream.filter(object -> object.isEmpty());
        SingleOutputStreamOperator<String> eventType = resultStream
                .filter(object -> JSON.parseObject(object)
                        .getString("eventType") != null);
        eventType.print();
        //4.数据输出-sink
        String[] columns = new String[]{
                "eventId", "cameraId", "licensePlate", "vehicleType", "roadId", "crossingId", "ts",
                "speed", "direction", "lane", "eventType", "areaId"
        };
        String upsertSql = "INSERT INTO traffic_monitoring.t_track_info (eventId, cameraId, licensePlate, vehicleType, roadId, crossingId, ts, " +
                "speed, direction, lane, eventType, areaId) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
        // 5. 存储违法车辆轨迹到 ClickHouse
        // 插入数据到ClickHouse
        JdbcUtil.sinkToClickhouseUpsertJson(eventType, columns, upsertSql);
        //5.执行-execute
        env.execute("IllegalVehicleTrackJob");
    }
}


